package com.hu.flink12.api

import java.sql.Types

import com.hu.util.YamlUtil
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.connector.jdbc.{JdbcInputFormat, JdbcOutputFormat}
import org.apache.flink.streaming.api.scala._

/**
 * @Author: hujianjun
 * @Create Date: 2021/1/28 13:44
 * @Describe:
 */
case class SensorReading(id: String, timeStamp: Long, temperature: Double)

object JdbcMysqlSink {
  def main(args: Array[String]): Unit = {
    val startTime = System.currentTimeMillis()

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val yamlUtil = YamlUtil.loadYaml("application.yml")

    //读mysql
    val dataSource = env.createInput(JdbcInputFormat.buildJdbcInputFormat()
      .setDrivername(YamlUtil.getProperty(yamlUtil, "dataSource.mysql.driver").toString)
      .setDBUrl(YamlUtil.getProperty(yamlUtil, "dataSource.adb.url").toString)
      .setUsername(YamlUtil.getProperty(yamlUtil, "dataSource.adb.userName").toString)
      .setPassword(YamlUtil.getProperty(yamlUtil, "dataSource.adb.passWord").toString)
      .setQuery("select cast(id as int) id,name email from iyb_clientele limit 100000")
//      .setFetchSize(1000)
      .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
      .finish
    )

    dataSource.writeUsingOutputFormat(JdbcOutputFormat
      .buildJdbcOutputFormat()
      .setDrivername(YamlUtil.getProperty(yamlUtil, "dataSource.mysql.driver").toString)
      .setDBUrl(YamlUtil.getProperty(yamlUtil, "dataSource.mysql.url").toString)
      .setUsername(YamlUtil.getProperty(yamlUtil, "dataSource.mysql.userName").toString)
      .setPassword(YamlUtil.getProperty(yamlUtil, "dataSource.mysql.passWord").toString)
      .setQuery("insert into test.user2 (id,email) values (?,?)")
      .setBatchSize(5000)
      .setSqlTypes(Array[Int](Types.INTEGER, Types.NCHAR))
      .finish)

    env.execute("flink batch sync mysql data!")
    println(s"total time is ${(System.currentTimeMillis() - startTime) / 1000}")
  }
}
